package task

import (
	"fmt"
	"time"
)

//写一个并发版本的Runner

type CurrentRunner struct {
	ResultChannel     dataChan
	ControllerChannel controllerChan
	ParamChannel      dataChan
	DataSize          int
	ConsumerNum       int
	ProducerNum       int
	Consumer          consumer
	Producer          producer
}

type consumer func(param dataChan, result dataChan, controller controllerChan) error

type producer func(param dataChan, controller controllerChan) error

func NewCurrentRunner(cf consumer, pf producer, c int, p int, size int) *CurrentRunner {
	return &CurrentRunner{
		ResultChannel:     make(chan interface{}, size),
		ControllerChannel: make(chan string, 1),
		ParamChannel:      make(chan interface{}, size),
		ConsumerNum:       c,
		ProducerNum:       p,
		Consumer:          cf,
		Producer:          pf,
	}
}
func (r *CurrentRunner) Run() {
	for i := 0; i < r.ProducerNum; i++ {
		fmt.Println("启动生产者线程")
		go r.Producer(r.ParamChannel, r.ControllerChannel)
	}
	for i := 0; i < r.ConsumerNum; i++ {
		fmt.Println("启动消费者者线程")
		go r.Consumer(r.ParamChannel, r.ResultChannel, r.ControllerChannel)
	}
}

func (r *CurrentRunner) Close() {
	i := r.ConsumerNum + r.ProducerNum
	for i != 0 {
		r.ControllerChannel <- "STOP"
		i--
	}
}

func Run() {
	var cf consumer = func(param dataChan, result dataChan, controller controllerChan) error {
		for {
			select {
			case p := <-param:
				//对P进行处理
				t := p.(string)
				fmt.Println("消费者接受数据: ",t)
				t = t + "已经处理"
				result <- t
			case <-controller:
				break
			default:

			}
		}
	}
	var pf producer = func(param dataChan, controller controllerChan) error {
		i := 0
		for {
			select {
			case <-controller:
				break
			default:
				var v = fmt.Sprintf("这是我生成的第 %d 个数据", i)
				fmt.Println("生产者生成数据: ",v)
				i++
				param <- v
			}
		}
	}
	t := time.Tick(time.Second * 10)

	runner := NewCurrentRunner(cf, pf, 2, 2, 4)
	runner.Run()
	fmt.Println("开始运行")
	for {
		select {
		case r := <-runner.ResultChannel:
			fmt.Println(r)
		case <-t:
			runner.Close()
			break
		default:
		}
	}
}
